#include "scheduler.h"
#include "hook.h"
#include "util.h"
#include "fiber.h"
#include "log.h"

#include <cstddef>
#include <functional>
#include <memory>
#include <mutex>
#include <thread>
#include <sstream>
#include <unistd.h>
#include <vector>

//协程调度器的指针
static thread_local Scheduler* t_scheduler = nullptr;
//协程调度器的调度协程
static thread_local Fiber* t_scheduler_fiber = nullptr;



Scheduler::Scheduler(size_t threads, bool use_caller, const std::string& name):m_name(name)
{
    ASSERT(threads > 0);

    //如果协程调度器线程数包含当前线程
    if(use_caller)
    {
        //如果当前线程没有协程，就创建一个主协程
        Fiber::GetThis();
        --threads;

        ASSERT(GetThis() == nullptr); //一个线程只能有一个协程调度器
        t_scheduler = this; //协程调度器的指针

        //创建协程调度器的调度协程
        m_rootFiber.reset(new Fiber(std::bind(&Scheduler::run, this), 0, false, "run"));

        t_scheduler_fiber = m_rootFiber.get();
        m_rootThread = GetThreadId(std::this_thread::get_id());
        m_threadIds.emplace_back(m_rootThread);
    }
    else 
    {
        m_rootThread = 0;
    }

    m_threadCount = threads;
}

Scheduler::~Scheduler()
{
    ASSERT(m_stopping); //协程调度器是否停止
    if(GetThis() == this)
    {
        t_scheduler = nullptr;
    }
}

//启动协程调度器
void Scheduler::start()
{
    std::unique_lock<std::mutex> lock(m_mutex);
    if(!m_stopping)
    {
        return;
    }
    m_stopping = false;

    ASSERT(m_threads.empty()); //断言线程池为空

    //给线程池添加线程
    m_threads.resize(m_threadCount);
    for(size_t i = 0; i < m_threadCount; ++i)
    {
        m_threads[i].reset(new std::thread(std::bind(&Scheduler::run, this)));
        m_threadIds.push_back(GetThreadId(m_threads[i]->get_id()));
        //c++ 提供了一个std::thread::native_handle() 函数来获取与特性线程库实现相关的handle,以此来提供更多线程控制能力
        // auto a = m_threads[i]->native_handle();
    }
}

//停止协程调度器
void Scheduler::stop()
{
    m_autoStop = true;
    if(m_rootFiber  //创建协程调度器时使用了当前线程
        && m_threadCount == 0
        && (m_rootFiber->getState() == Fiber::TERM 
            || m_rootFiber->getState() == Fiber::INIT)) //调度协程的状态为结束或初始化
    {
        log_info("Scheduler stopped");
        m_stopping = true;

        if(stopping())
        {
            return;
        }
    }

    if(m_rootThread != 0)   //创建协程调度器时使用了当前线程
    {
        ASSERT(GetThis() == this); //目的确保stop函数是在创建协程调度器的线程上执行的
    }
    else
    {
        ASSERT(GetThis() != this);
    }
    
    m_stopping = true;
    
    //唤醒线程池中所有的线程，使其执行完自动结束
    for(size_t i = 0; i < m_threadCount; ++i)
    {
        tickle();
    }

    if(m_rootFiber)
    {
        tickle();   //确保每一个线程都执行tickle()
        if(!stopping())
        {
            m_rootFiber->call();
        }
    }

    std::vector<std::shared_ptr<std::thread>> thrs;
    {
        std::unique_lock<std::mutex> lock(m_mutex);
        thrs.swap(m_threads);
    }

    for(auto& t: thrs)
    {
        t->join();
    }
}

//返回当前协程调度器
Scheduler* Scheduler::GetThis()
{
    return t_scheduler;
}

//返回当前协程调度器的调度协程
Fiber* Scheduler::GetMainFiber()
{
    return t_scheduler_fiber;
}

//获取线程id
uint64_t Scheduler::GetThreadId(std::thread::id id)
{
    std::ostringstream oss;
    oss << id;
    std::string idString = oss.str();

    return std::stoull(idString);
}

//通知协程调度器有任务了
void Scheduler::tickle()
{
    log_info("Scheduler::tickle  thread id = {}", GetThreadId(std::this_thread::get_id()));
}
    
//返回是否可以停止
bool Scheduler::stopping()
{
    std::unique_lock<std::mutex> lock(m_mutex);
    return m_autoStop                //是否自动停止
        && m_stopping                //当前是否停止
        && m_fibers.empty()          //任务队列为0   
        && m_activeThreadCount == 0; //活跃线程数为0
}

//协程无任务时可调度执行idle协程
void Scheduler::idle()
{
    log_info("Scheduler::idle thread id = {}", GetThreadId(std::this_thread::get_id()));
    while(!stopping())
    {
        sleep(1);
        Fiber::YieldToHold();
    }
}

//协程调度函数
void Scheduler::run()
{
    log_info("Scheduler name:{} run thread id = {}", m_name, GetThreadId(std::this_thread::get_id()));

    set_hook_enable(true); //每一个协程调度器的线程都要使用协程
    setThis(); //设置当前线程属于哪个协程调度器

    //如果当前线程不是use_caller线程
    if(GetThreadId(std::this_thread::get_id()) != m_rootThread)
    {
        t_scheduler_fiber = Fiber::GetThis().get(); //给当前线程创建主协程
    }   

    //创建idle协程，当调度器没有任务时执行
    Fiber::ptr idle_fiber(new Fiber(std::bind(&Scheduler::idle, this), 0, true, "idle"));
    
    Fiber::ptr cb_fiber;
    FiberAndThread ft;

    while(true)
    {
        ft.resetState();
        bool tickle_me = false; //是否用其他线程执行的任务
        bool is_active = false; //是否空闲(当前是否有任务)

        {   //取出待执行的任务
            std::unique_lock<std::mutex> lock(m_mutex);

            auto it = m_fibers.begin();
            while(it != m_fibers.end())
            {
                //如果指定任务的执行的线程不为当前线程
                if(it->threadId != 0 && it->threadId != GetThreadId(std::this_thread::get_id()))
                {
                    ++it;
                    tickle_me = true;
                    continue;
                }

                ASSERT(it->fiber || it->cb);
                //当前协程为执行中时跳过该协程
                if(it->fiber && it->fiber->getState() == Fiber::EXEC)
                {
                    ++it;
                    continue;
                }

                ft = *it;
                m_fibers.erase(it++);
                ++m_activeThreadCount;
                is_active = true;
                break;
            }
        }

        //通知其他线程处理任务
        if(tickle_me) 
        {
            tickle();
        }

        //任务为协程
        if(ft.fiber && (ft.fiber->getState() != Fiber::TERM //不为结束
                        && ft.fiber->getState() != Fiber::EXCEPT))  //不为异常
        {
            ft.fiber->swapIn(); //将当前调度协程挂起，执行任务协程
            --m_activeThreadCount;

            //协程切换回来之后
            if(ft.fiber->getState() == Fiber::READY)
            {
                schedule(ft.fiber);
            }
            else if(ft.fiber->getState() != Fiber::TERM     //不为结束
                    && ft.fiber->getState() != Fiber::EXCEPT)   //不为异常  说明当前协程状态为执行中
            {
                ft.fiber->m_state = Fiber::HOLD; // 暂停
                // schedule(ft.fiber);
            }

            ft.resetState();
        }
        else if(ft.cb)  //任务为function
        {
            if(cb_fiber)
            {
                cb_fiber->resetFunc(ft.cb); //重置协程执行函数
            }
            else
            {
                cb_fiber.reset(new Fiber(ft.cb, 0, true, "任务"));
            }
            ft.resetState();

            cb_fiber->swapIn();
            --m_activeThreadCount;
            if(cb_fiber->getState() == Fiber::READY)
            {
                schedule(cb_fiber);
                cb_fiber.reset();
            }
            else if(cb_fiber->getState() == Fiber::EXCEPT    //异常
                    || cb_fiber->getState() == Fiber::TERM)  //结束
            {
                cb_fiber->resetFunc(nullptr);
            }
            else    //HOLD 或者 EXEC
            {//if(cb_fiber->getState() != Fiber::TERM) {
                cb_fiber->m_state = Fiber::HOLD; // 暂停
                // schedule(cb_fiber);
                cb_fiber.reset(); 
            }
        }
        else    //当前已经没有任务了 
        {
            if(is_active) //如果当前还有任务
            {
                --m_activeThreadCount;
                continue;
            }
            if(idle_fiber->getState() == Fiber::TERM)
            {
                log_info("idle fiber term, thread id = {}", GetThreadId(std::this_thread::get_id()));
                break;
            }
            
            ++m_idleThreadCount;
            idle_fiber->swapIn();
            --m_idleThreadCount;
            if(idle_fiber->getState() != Fiber::TERM && idle_fiber->getState() != Fiber::EXCEPT)
            {
                idle_fiber->m_state = Fiber::HOLD;
            }
        }
    }
}

//设置当前的协程调度器
void Scheduler::setThis()
{
    t_scheduler = this;  
}